package P3;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.Vector;

import NIOCommunication.NIOProcessor;
import NIOCommunication.NIOProcessorCloseAllCommand;
import NIOCommunication.NIOProcessorCloseAllExceptCommand;
import NIOCommunication.NIOSocketContext;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.util.Iterator;

public class Worker extends Thread
{
    public enum State
    {
        STATE_WAITING_FOR_COORDINATOR,
        STATE_WAITING_TO_START,
        STATE_CRAWLING,
        STATE_SENDING_DATA,
        STATE_ALL_DATA_SENT,
        STATE_DONE,
    }
    
	//private LinkedHashSet<Peer> mPeers = new LinkedHashSet<Peer>();
	private HashSet<Long> mCrawledPeers = new HashSet<Long>();
	private HashSet<Long> mUncrawledPeers = new HashSet<Long>();
        private boolean mMinimal = true;
        private CoordinatorInterface mCoordinatorConnection;
        private NIOProcessor mProcessor;
        private State mState = State.STATE_WAITING_FOR_COORDINATOR;
        private boolean mClosed = false;
        private FileWriter mLog;
	
        public Worker()
        {
            mProcessor = new NIOProcessor();
            try
            {
                File log = new File("Worker.log");
                if(log.exists())
                {
                    log.delete();
                }
                mLog = new FileWriter(log);
            }
            catch(Exception e)
            {
                e.printStackTrace();
            }
        }
        
        public void OnPeersGathered(CrawlResult result)
        {
            if(mState != State.STATE_CRAWLING)
            {
                return;
            }
            
            if(!mMinimal && mState == State.STATE_CRAWLING)
            {
                String[] tokens = result.Address.split(":");
                try
                {
                    new FileGathererContext(mProcessor, new InetSocketAddress(tokens[0], Integer.parseInt(tokens[1])),result);//, result);
                    System.out.println(result.FancyToString());
                }
                catch(Exception e)
                {
                    e.printStackTrace();
                }
            }
            else
            {
                OnFilesGathered(result);
            }
        }
        
        int mNumResults = 0;
        public void OnFilesGathered(CrawlResult result)
        {
            // what do we want to do with the result???  for now let's send it to the coordinator and parse the peers list
            //System.out.println(result.toString());
            
            ++mNumResults;
            
            //int crawled = mCrawledPeers.size();
            //int uncrawled = mUncrawledPeers.size();
            //System.out.println("Finished crawling node " + result.Address);
            //System.out.println(mNumResults + " Results after crawling " + crawled + " out of " + (crawled+uncrawled) + " known peers.  Uncrawled = " + uncrawled);
	
            String peersLine = new String(result.Peers);
            if(peersLine.trim().length() == 0)
            {
                //System.out.println("Found no peers");
            }
            else
            {
                String[] peers = peersLine.split(",");
		OnNewPeers(null, peers, true);
            }
            
            // TODO: Save results until the server asks for them
            //SendMessage(result.toString());
            try
            {
                mLog.write(result + "\n");
            }
            catch(Exception e)
            {
                e.printStackTrace();
            }
        }
        
	public void OnNewPeers(InetSocketAddress source, String[] peers, boolean isUltraPeer)
	{
		for(int i = 0; i < peers.length; ++i)
		{
			//mPeers.add(new Long(AddressConverter.GetBytes(peers[i])));
			mUncrawledPeers.add(new Long(AddressConverter.GetBytes(peers[i])));
			//Peer newPeer = new Peer(peers[i], isUltraPeer);
		//	mPeers.add(newPeer);	
		}
                if(mState == State.STATE_CRAWLING)
                {
                    CrawlNewPeer();
                }
	}
	
	public void CrawlNewPeer()
	{
		Iterator<Long> it = mUncrawledPeers.iterator();
		while(it.hasNext())
		{
			if(!NIOSocketContext.IsSocketAvailable())
			{
				return;
			}
			Long peer = it.next();
                        //System.out.print("Peer: " + AddressConverter.GetString(peer.longValue()));
			new PeerGathererContext(this, mProcessor, new InetSocketAddress(AddressConverter.GetHostString(peer), AddressConverter.GetPort(peer)), peer.longValue());
			mCrawledPeers.add(peer);
			it.remove();
		}
	}
        
        public void Close()
        {
            System.out.println("No more messing around.  Let's just QUIT!");
            System.exit(0);
            
            if(!mClosed)
            {
                mClosed = true;
                mState = State.STATE_DONE;
                System.out.println("Worker closed");
                mProcessor.Kill();
                if(mLog != null)
                {
                    try
                    {
                        mLog.close();
                    }
                    catch(Exception e)
                    {
                        e.printStackTrace();
                    }
                }
            }
        }
        
        public void OnNewMessage(String message)
        {
            // we've received a message from the coordinator.
            System.out.println("New Message: " + message);
            if(message.startsWith("Bootstrap\n"))
            {
                try
                {
                    if(mState != State.STATE_WAITING_TO_START)
                    {
                        System.out.println("Told to bootstrap when I wasn't waiting to start!");
                        Close();
                        return;
                    }
                    
                    String bootstrapNode = message.substring(message.indexOf("\n") + 1);
                    String[] tokens = bootstrapNode.split(":");
                    String HostName = tokens[0];
                    int Port = Integer.parseInt(tokens[1]);
                    
                    mState = State.STATE_CRAWLING;
                    InetAddress address = InetAddress.getByName(HostName);
                    String peerName = address.getHostAddress() + ":" + Port;
                    Long peer = AddressConverter.GetBytes(peerName);
                    new PeerGathererContext(this, mProcessor, new InetSocketAddress(address, Port), peer);
                    //new PeerGathererContext(this, mNIOProcessor, new InetSocketAddress(InetAddress.getByName("reala.ece.ubc.ca"), 5627));
                    //new PeerGathererContext(this, mNIOProcessor, new InetSocketAddress(InetAddress.getByName("24.94.248.15"), 51778));

                	//new PeerGathererContext(this, mNIOProcessor, new InetSocketAddress(InetAddress.getByName("86.52.78.16"), 36822));
                    //System.out.println("starting file gathering");
                	//new FileGathererContext(mProcessor, new InetSocketAddress(InetAddress.getByName("68.100.6.63"),15958));

                }
                catch(Exception e)
                {
                    e.printStackTrace();
                }
            }
            if(message.equals("Send Data"))
            {
                if(mState != State.STATE_CRAWLING)
                {
                    System.out.println("Told to send data when I wasn't crawling!");
                    Close();
                    return;
                }
                mState = State.STATE_SENDING_DATA;
            }
        }
        
        public void SendData()
        {
            if(mState != State.STATE_SENDING_DATA)
            {
                Close();
                return;
            }
            // TODO: send data back.
            mProcessor.QueueCommand(new NIOProcessorCloseAllExceptCommand(mCoordinatorConnection));
            System.out.println("Worker: Sending data");
            System.out.println("Worker: Sending data");
            System.out.println("Worker: Sending data");
            System.out.println("Worker: Sending data");
            System.out.println("Worker: Sending data");
            System.out.println("Worker: Sending data");
            
            FileReader fr = null;
            try
            {
                fr = new FileReader(new File("Worker.log"));
                BufferedReader in = new BufferedReader(fr);
        
                CrawlResult result = null;
                StringBuilder text = null;
                String line = null;
                boolean done = false;
                while(!done)
                {
                    text = new StringBuilder();
                    result = new CrawlResult();
                    for(int i = 0; i < 9; ++i)
                    {
                        line = in.readLine();
                        if(line == null)
                        {
                            mProcessor.QueueCommand(new NIOProcessorCloseAllExceptCommand(mCoordinatorConnection));
                            done = true;
                        }
                        text.append(line + "\n");
                    }
                    if(!done)
                    {
                        result.LoadFromString(text.toString());
                        mCoordinatorConnection.SendMessage(result.toString());
                    }
                }   
            }
            catch(Exception e)
            {
                e.printStackTrace();
            }
            
            System.out.println("Sending the uncrawled");
            System.out.println("Sending the uncrawled");
            System.out.println("Sending the uncrawled");
            System.out.println("Sending the uncrawled");
            System.out.println("Sending the uncrawled");
            System.out.println("Sending the uncrawled");
            System.out.println("Sending the uncrawled");
            System.out.println("Sending the uncrawled");
            System.out.println("Sending the uncrawled");
            CrawlResult result = new CrawlResult();
            result.Status = "Uncrawled";
            Iterator<Long> it = mUncrawledPeers.iterator();
            while(it.hasNext())
            {
                result.Address = AddressConverter.GetString(it.next().longValue());
                mCoordinatorConnection.SendMessage(result.toString());
            }
            
            mCoordinatorConnection.SendMessage("Goodbye");
            mState = State.STATE_ALL_DATA_SENT;
            System.out.println("Done sending the uncrawled");
            System.out.println("Done sending the uncrawled");
            System.out.println("Done sending the uncrawled");
            System.out.println("Done sending the uncrawled");
            System.out.println("Done sending the uncrawled");
            System.out.println("Done sending the uncrawled");
            System.out.println("Done sending the uncrawled");
              
            //mProcessor.CloseAllExcept(mCoordinatorConnection);
            //mProcessor.QueueCommand(new NIOProcessorCloseAllExceptCommand(mCoordinatorConnection));
            //mCoordinatorConnection.SendMessage("Here is my data!");
        }
        
        public void WaitForCoordinationWrites()
        {
            try
            {
                while((mState != State.STATE_ALL_DATA_SENT) || mCoordinatorConnection.IsWritePending())
                {
                    System.out.println("Waiting for coordination writes to finish");
                    Thread.sleep(1000);
                }
            }
            catch(Exception e)
            {
                e.printStackTrace();
            }
        }
	
	public void OnNewFiles(InetSocketAddress source, String[] file)
	{
		
	}
        
        public void OnCrawlerTimedOut()
        {
            System.out.println("Connection Timed Out");
        }
	
	public void OnCrawlerClosed()
	{
            if(mState == State.STATE_CRAWLING)
		CrawlNewPeer();
	}
        
        public void OnCoordinatorDisconnected(CoordinatorInterface coord)
        {
            System.out.println("Coordination channel closed");
            if(mState != State.STATE_WAITING_FOR_COORDINATOR)
            {
                 Close();   
            }
        }
        
        public void OnCoordinatorTimedout(CoordinatorInterface coord)
        {
            System.out.println("Coordination channel timed out");
        }
        
        public void WaitForCoordinatorConnection(InetSocketAddress address)
        {
            try
            {
                System.out.println("Waiting for coordinator...");
                mState = State.STATE_WAITING_FOR_COORDINATOR;
                while(true)
                {
                    System.out.println("Trying to create the connection");
                    mCoordinatorConnection = new CoordinatorInterface(this, mProcessor, address);
                    
                    
                    //Thread.sleep(1000);
                    //Close();
                    
                    while(true)
                    {
                        if(mCoordinatorConnection.IsConnected()) // Success!
                        {
                            mState = State.STATE_WAITING_TO_START;
                            System.out.println("Connected to coordinator!");
                            mCoordinatorConnection.SendMessage("Hello Coordinator!");
                            return;
                        }
                        if(mCoordinatorConnection.IsClosed()) // No luck.
                        {
                            System.out.println("Connection was closed...");
                            break;
                        }
                        if(mProcessor.IsClosed())
                        {
                            System.out.println("Processor was stopped");
                            mState = State.STATE_DONE;
                            break;
                        }
                        
                        mProcessor.EnsureThreadIsRunning();
                    
                        Thread.sleep(500);
                    }
                }
                
            }
            catch(Exception e)
            {
                e.printStackTrace();
            }
            
            return;
        }
	
    public static void main(String[] args)
    {	
    	String coordHostAddress = null;
    	int coordHostPort = 0;
    	int tcpTimeout = 0;
    	
    	
		Vector<String> cooradd = new Vector();
		try{
/*
 * For Reading in a input file		
 */
//			String thisLine = null;
//			BufferedReader reader = new BufferedReader(new FileReader("COORDHOSTLIST.txt"));
//			while ((thisLine = reader.readLine()) != null) {
//
//			} 

    		coordHostAddress = (args[0].split(":"))[0];
    		coordHostPort = Integer.parseInt((args[0].split(":"))[1]);
    		tcpTimeout = Integer.parseInt(args[1]);
    		
    		
    	}catch(NumberFormatException e){
    		coordHostPort = 9876;
//    	}catch(FileNotFoundException e){
//    		
//    	}catch(IOException e){
//    		
    	}catch(Exception e){
    		System.err.println(e);
    	}
    		
        try
        {
            InetSocketAddress coordinatorAddress = new InetSocketAddress(coordHostAddress, coordHostPort);
            System.out.println(coordinatorAddress.getAddress()+":"+coordinatorAddress.getPort());
            while(true)
            {
                Worker worker = new Worker();
                worker.WaitForCoordinatorConnection(coordinatorAddress);
                while(worker.mState != State.STATE_SENDING_DATA && worker.mState != State.STATE_DONE)
                {
                    try
                    {
                        if(worker.mState == State.STATE_WAITING_TO_START)
                        {
                            System.out.println("Worker: Waiting for the coordinator to start us.");
                        }
                        else
                        {
                            System.out.println("Worker: Waiting for the coordinator to stop us.");
                        }
                        //if(worker.IsClosed())
                        //{
                        //    System.out.println("Coordination channel closed!  Current crawl aborted.");
                        //    processor.QueueCommand(new NIOProcessorCloseAllCommand());
                        //    break;
                        //}
                        Thread.sleep(1000);
                    }
                    catch(Exception e)
                    {
                        e.printStackTrace();
                    }
                }
                
                if(worker.mState == State.STATE_DONE)
                {
                    worker.Close();
                }
                
                if(worker.mState == State.STATE_SENDING_DATA)
                {
                    System.out.println("Starting to send data");
                    worker.SendData();
                    worker.WaitForCoordinationWrites();
                    System.out.println("All writes finished.  Closing worker");
                }
                worker.Close();
                
                Thread.sleep(10000); // just to be safe
            }
        }
        catch (Exception e)
        {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}
